package com.gitee.sparta.core.process.feature;

import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.common.IdentityEnum;
import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.cluster.AbstractClusterFollower;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 智能睡眠特性
 *
 * <p>
 * 读取 timeout_tab 的线程，根据上一次读取结果中返回的 timeout_tab中剩
 * 余最小的 score 的值来计算本次将要睡眠多久
 *
 * <p>
 * 例如：
 * minScore： 1500
 * current timestamp：1000 （System.currentTimeMillis() 获取的值）
 * 最终要睡眠的时间：minScore - currentTimestamp =>  1500 - 1000 = 500ms
 *
 * <p>
 * 假如你在担心分布式部署的情况下，相同的项目被部署了两台机器，其中有一台先除
 * 了掉了本台机器读取的 minScore 的值，会导致本台机器读取的内容延迟的问题
 * 当然，这种情况是不用担心的，在本系统中，是不允许插入一条比当前时间戳还小
 * 的任务进去的，即便是 minScore 被别的机器消费了，大不了本台机器空转一次
 *
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class SpartaWorkerFeature {

    /** 关机钩子函数中，正在关闭整个服务，睡眠中的 worker 线程强行被打断后会发送 candidate 消息 **/
    public static volatile boolean FORBID_CANDIDATE = false;

    private static long prev = 0;

    /** 没有找到剩余的任务所返回的值 **/
    private static final int NOT_FOUND = 0;
    /** 当前时间戳与最小的分数的差值位 30ms 的时候，则不睡眠 **/
    private static final int MIN_SCOPE = 5;

    /** 这里保存当前线程的目的，是为了任何时候需要打断搬运线程睡眠都是可以的 */
    private volatile static Thread SPARTA_READ_THREAD = null;
    /** reentrantLock **/
    private static final ReentrantLock LOCK = new ReentrantLock(false);

    /**
     * 调用地方：
     *  1. 插入数据（N）个线程
     *  2. 重试或续期 (N) 个线程
     * <p>
     * 打断正在睡眠中的读取 timeout_tab 的线程
     * 方法调用时机：当有任务被添加时，该方法主动调用
     * <p>
     */
    public static void interruptWorker () {
        long timeout = 5L;
        // 如果自己是 follower 节点
        if (Objects.equals(IdentityEnum.FOLLOWER, ClusterCache.getServerIdentity())) {
            long exp = System.currentTimeMillis() - prev;
            // 间隔 1000ms 发送一次
            long pause = 1000;
            if (exp > pause) {
                ClusterProtocol protocol = new ClusterProtocol(
                        ClusterCmd.INTERCEPTOR,
                        ClusterCache.getServerId(),
                        ClusterCache.getLeaderSid()
                );
                AbstractClusterFollower.sendQueue(protocol);
                prev = System.currentTimeMillis();
            }
            return;
        }
        if (Objects.isNull(SPARTA_READ_THREAD)) {
            return;
        }
        try {
            // 抢占锁，如果在 5ms 内没有抢占到锁，那么就直接退出
            // 如果执行的线程被中断过，那么就抛出中断异常
            if (LOCK.tryLock(timeout, TimeUnit.MILLISECONDS) && Objects.nonNull(SPARTA_READ_THREAD)) {
                LockSupport.unpark(SPARTA_READ_THREAD);
                SPARTA_READ_THREAD = null;
            }
        } catch (InterruptedException e) {
            log.error("[Sparta] Interrupt worker, current thread {} interrupted, error : {}",
                    Thread.currentThread().getName(), e.getMessage(), e);
        } finally {
            LOCK.unlock();
        }
    }

    /**
     * 读取 timeout_tab 的线程进入睡眠状态
     *
     * @param minScore    timeout_tab 列表内最小的分数
     */
    public static void nextWorker (long minScore) {
        SPARTA_READ_THREAD = Thread.currentThread();
        // 读取到无数据的情况，睡眠最大时间 10s
        if (minScore == NOT_FOUND) {
            LockSupport.parkNanos(SpartaConstant.MAX_SLEEP_SEC * SpartaConstant.NANOS);

            if (FORBID_CANDIDATE) {
                AbstractClusterLeader.sendQueue(ClusterProtocol.defaultValue(
                        ClusterCmd.FLUSH_SERVER_TABLE
                ));
            }
            SPARTA_READ_THREAD = null;
            return;
        }
        // 消息已经过期
        if (System.currentTimeMillis() >= minScore) {
            SPARTA_READ_THREAD = null;
            return;
        }
        long sleepTime = minScore - System.currentTimeMillis();
        // 剩余还有时间， 但是不超过 30ms
        if (sleepTime <= MIN_SCOPE) {
            SPARTA_READ_THREAD = null;
            return;
        }
        // 消息过期时间超过 30ms
        LockSupport.parkNanos(sleepTime * SpartaConstant.NANOS_MILLISECONDS);
        SPARTA_READ_THREAD = null;
    }
}
